package com.ht.interfaces.kafka;

import java.util.List;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.context.annotation.Profile;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import com.ht.api.db.Db;
import com.ht.api.db.DbConfig;
import com.ht.api.db.Entity;
import com.ht.interfaces.constant.MqConstants;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;

/**
 * kafka消费者
 * @author asq
 * @createTime 2024年12月7日 10:22:22
 */
@Slf4j
@Component
@Profile("prod")
public class KafkaConsumer {
	@KafkaListener(topicPattern = MqConstants.TOPIC_PRE_DATA + ".*", containerFactory = "kafkaListenerContainerFactory")
	public void consumeBatchByData(List<ConsumerRecord<String, String>> records, Acknowledgment ack) {
		log.info("[kafka]收到entity数据批量消息：size:{}条", records.size());

		records.forEach(record -> {
			try {
				String topic = record.topic();
				long tenantId = Long.parseLong(topic.substring(MqConstants.TOPIC_PRE_DATA.length()));
				String message = record.value();

				// 解析消息中的data字段
				JSONObject messageObj = JSONUtil.parseObj(message);
				String dataStr = messageObj.getStr("data");
				List<Entity> list = JSONUtil.toList(dataStr, Entity.class);
				
				// 处理消息
				Db.insertBatchWithStream(DbConfig.useTenantSLC(tenantId), list);				
				log.info("[kafka]处理entity数据成功，租户ID：{}, 数据条数：{}", tenantId, list.size());
			} catch (Exception e) {
				log.error("[kafka]处理消息失败，错误：{}", e.getMessage(), e);
			}
		});

		// 手动提交偏移量
		ack.acknowledge();
	}
}